package com.cn.file.index;

import com.cn.file.DispatchRequest;
import com.cn.file.DispatcherService;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;

import java.io.File;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class IndexService implements DispatcherService {

    private final ReentrantReadWriteLock writeLock = new ReentrantReadWriteLock();
    private String path;
    private List<IndexFile> listIndex = new ArrayList<>();
    private int hashSlotNum = 5000000;//hash槽的数目
    private int indexCount = 5000000 * 4;//索引总数
    private ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    public IndexService(String path) {
        this.path = path;
    }

    public void load() {
        File files = new File(path);
        if (!files.exists()) {
            files.mkdirs();
        }
        System.out.println(path);
        List<File> listFile = new ArrayList<>();
        for (File file : files.listFiles()) {
            listFile.add(file);
        }
        Collections.sort(listFile);
        for (File f : listFile) {
            IndexFile indexFile = new IndexFile(f.getAbsolutePath(), hashSlotNum, indexCount);
            System.out.println("消息总数：" + indexFile.getIndexHeader().getIndexCount());
            listIndex.add(indexFile);
        }
    }

    @Override
    public void dispatch(DispatchRequest dispatchRequest) {
        putKey(dispatchRequest);
    }

    /**
     * 提交index
     *
     * @param dispatchRequest
     */
    public void putKey(DispatchRequest dispatchRequest) {
        this.writeLock.writeLock().lock();
        try {
            IndexFile indexFile = getIndexFile();
            if (indexFile.getIndexHeader().getEndPhyOffset().get() >= dispatchRequest.getCommitLogOffset()) {
                return;
            }
            if (StringUtils.isEmpty(dispatchRequest.getTopic())) {
                return;
            }
            indexFile.putKey(dispatchRequest.getTopic() + "@" + dispatchRequest.getQueueId(), dispatchRequest.getCommitLogOffset());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            this.writeLock.writeLock().unlock();
        }
    }

    public List<Long> selectIndex(String key) {
        IndexFile indexFile = getIndexFile();
        List<Long> result = new ArrayList<>();
        return indexFile.selectPhyOffset(result, key);
    }

    public IndexFile getIndexFile() {
        IndexFile indexFile;
        if (listIndex.isEmpty()) {
            //新建一个
            indexFile = createIndexFile();
        } else {
            indexFile = listIndex.get(listIndex.size() - 1);
            if (indexFile.getIndexHeader().getIndexCount() >= indexCount) {
                IndexFile old = indexFile;
                //新建一个
                indexFile = createIndexFile();

                //将上一个刷盘
                submitFlush(old, 0);
            }
        }
        return indexFile;
    }

    public void submitFlush(IndexFile indexFile, int count) {
        scheduledExecutorService.schedule(new Runnable() {
            @Override
            public void run() {
                flush(indexFile, count);
            }
        }, 500, TimeUnit.MILLISECONDS);
    }

    public void flush(IndexFile indexFile, int count) {
        try {
            indexFile.flush();
        } catch (Exception e) {
            if (count <= 5) {
                submitFlush(indexFile, count);
            }
        } finally {
            count++;
        }
    }

    public IndexFile createIndexFile() {
        String fileName = path + "/" + DateFormatUtils.format(System.currentTimeMillis(), "yyyyMMddHHmmss");
        IndexFile indexFile = new IndexFile(fileName, hashSlotNum, indexCount);
        listIndex.add(indexFile);
        return indexFile;
    }

    public static void main(String[] args) {
        IndexService indexService = new IndexService("C:\\F\\data3\\data\\rocketmq\\index");
        indexService.load();
        DispatchRequest dispatchRequest = new DispatchRequest("topic", 0, 100, 0, 0, 0);
//        indexService.dispatch(dispatchRequest);
        System.out.println(indexService.selectIndex("topic@0"));
    }

}
